package demo.kafka.spring;

import demo.utils.ThreadUtil;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ExecutionException;

public class SpringProducerDemo {

    private static Logger logger = LoggerFactory.getLogger(SpringProducerDemo.class);

    String topic4 = "realtime-security-access-event";
    private String servers = "192.168.3.68:9092,192.168.3.69:9092,192.168.3.70:9092";
    private int retries = 3;
    private int batchSize = 1;
    private int linger = 0;
    private int bufferMemory = 1 * 1024 * 1024;

    public Map<String, Object> producerConfigs() {
        Map<String, Object> map = new HashMap<>();
        /**
         * 用于建立与kafka集群连接的host/port组。 数据将会在所有servers上均衡加载，不管哪些server是指定用于bootstrapping。
         * 这个列表仅仅影响初始化的hosts（用于发现全部的servers） 这个列表格式：host1:port1,host2:port2,…
         */
        map.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        /**
         * 此配置是表明当一次发送消息请求被认为完成时的确认值，就是指procuder需要多少个broker返回的确认信号。此配置实际上代表了数据备份的可用性。以下设置为常用选项：
         * <li>（1）acks=0： 设置为0表示producer不需要等待任何确认收到的信息。副本将立即加到socket
         * buffer并认为已经发送。没有任何保障可以保证此种情况下server已经成功接收数据，同时重试配置不会发生作用（因为客户端不知道是否失败）回馈的offset会总是设置为-1；
         * <li>（2）acks=1：
         * 这意味着至少要等待leader已经成功将数据写入本地log，但是并没有等待所有follower是否成功写入。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。
         * <li>（3）acks=all： 这意味着leader需要等待所有备份都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的保证。
         * <li>（4）其他的设置，例如acks=2也是可以的，这将需要给定的acks数量，但是这种策略一般很少用。
         */
        map.put(ProducerConfig.ACKS_CONFIG, "all");
        /**
         * 如果设置de值大于0,客户端将重新发送之前发送失败的数据。 注意，这些重试与客户端接收到发送错误时的重试没有什么不同。
         * 在不将max.in.flight.requests.per.connection设置为1的情况下允许重试将潜在的改变数据的顺序，
         * 如果这两个消息记录都是发送到同一个partition，则第一个消息失败第二个发送成功，则第二条消息会比第一条消息出现要早。
         */
        map.put(ProducerConfig.RETRIES_CONFIG, retries);
        /**
         * 只要有多个记录被发送到同一个分区，生产者就会尝试将记录组合成一个batch的请求。 这有助于客户端和服务器的性能。
         * 该配置以字节为单位控制默认的批量大小。不要尝试大于次值的批量记录。发送给broker的请求将包含多个批次。
         * 太小的批量会降低吞吐量（批量为0讲完全禁用批处理），太大的批量会浪费一些内存，因为这批量的大小是预分配的。
         */
        map.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        /**
         * Producer默认会把两次发送时间间隔内收集到的所有请求进行一次汇总然后再发送，以此提高吞吐量，而linger.ms则更进一步，这个参数为每次发送增加一些延迟，以此来聚合更多的Message。
         * 通常来说，这只有在记录产生速度大于发送速度的时候才能发生。一旦汇集的消息数据达到了某个partition的batch.size，他将会立即发送而不顾这项设置，
         * 然而如果我们获得消息字节数比batch.size设置要小的多，我们需要“linger”特定的时间以获取更多的消息。
         * 这个设置默认为0，即没有延迟。设定linger.ms=5，例如，将会减少请求数目，但是同时会增加5ms的延迟。
         */
        map.put(ProducerConfig.LINGER_MS_CONFIG, Optional.of(linger));

        /**
         * producer用于压缩数据的压缩类型。默认是无压缩。正确的选项值是none、gzip、snappy和lz4。压缩最好用于批量处理，批量处理消息越多，压缩性能越好。
         * 推荐配置一种适合的压缩算法，可以大幅度的减缓网络压力和Broker的存储压力。 string none high
         */
        map.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");
        /**
         * 该配置控制KafkaProducer.send（）和KafkaProducer.partitionsFor（）将被阻塞多长时间。这些方法可能因为缓冲区已满或元数据不可用而被阻塞。用户提供的序列化程序或分区程序中的阻塞将不计入此超时。
         */
        map.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, Optional.of(60000));
        /**
         * 生产者可用于缓冲等待发送到服务器的记录的总字节数。
         * 如果记录的发送速度比发送到服务器的速度快，那么生产者将会阻塞max.block.ms，之后它将抛出一个异常。
         * 这项设置将和producer能够使用的总内存相关，但并不是一个硬性的限制，因为不是producer使用的所有内存都是用于缓存。
         * 一些额外的内存会用于压缩（如果引入压缩机制），同样还有一些用于维护请求。
         */
        map.put(ProducerConfig.BUFFER_MEMORY_CONFIG, Optional.of(bufferMemory));

        /** 空闲连接的超时关闭时间 */
        map.put(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG, Optional.of(540000));
        /** 关键字的序列化类。如果没给与这项，默认情况是和消息一致 */
        map.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        /** 值的序列化类 */
        map.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return map;
    }

    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    static String msg = "醉里挑灯看剑，梦回吹角连营。八百里分麾下炙，五十弦翻塞外声。沙场秋点兵。马作的卢飞快，弓如霹雳弦惊。了却君王天下事，赢得生前身后名。可怜白发生！";

    public static void main(String[] args) throws Exception {
        System.err.println(msg.getBytes(StandardCharsets.UTF_8).length);
        SpringProducerDemo demo = new SpringProducerDemo();
        KafkaTemplate<String, String> kafkaTemplate = demo.kafkaTemplate();
        int count = 10;
        {
            long t1 = System.currentTimeMillis();
            for (int i = 0; i < count; i++) {
                demo.sendAsync(kafkaTemplate);
            }
            long t2 = System.currentTimeMillis();
            logger.info("异步发送{}次，耗时：{} ms.", count, (t2 - t1));
        }
        {
            long t0 = System.currentTimeMillis();
            for (int i = 0; i < count; i++) {
                demo.sendSync(kafkaTemplate);
            }
            long t1 = System.currentTimeMillis();
            logger.info("同步发送{}次，耗时：{} ms.", count, (t1 - t0));
        }
        ThreadUtil.safeSleep(30 * 1000L);
    }

    void sendSync(KafkaTemplate<String, String> kafkaTemplate) throws InterruptedException, ExecutionException {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic4, "key1", msg);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                logger.error("同步发送KAFKA消息失败", throwable);
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                logger.info("同步发送KAFKA消息成功：{}", result);
            }
        });
        future.get();
    }

    void sendAsync(KafkaTemplate<String, String> kafkaTemplate) {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic4, "key1", msg);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                logger.error("异步发送KAFKA消息失败", throwable);
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                logger.info("异步发送KAFKA消息成功：{}", result);
            }
        });
    }
}
